/****************/
/* Your Name	*/
/* Date		*/
/* CS 244B	*/
/* Spring 1994	*/
/****************/

#define DEBUG

#include <fcntl.h>
#include <sys/stat.h>

#include <sys/types.h>
#include <sys/uio.h>
#include <sys/socket.h>
#include <sys/time.h>

#include <netinet/in.h>
#include <arpa/inet.h>

#include <unistd.h>
#include <errno.h>
#include <netdb.h>
#include <signal.h>
#include <pwd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <assert.h>

#include <client.h>


/* Types */
typedef	struct sockaddr_in		Sockaddr;

/* Constants */
#define TRUE					1
#define FALSE					0
#define MaxWBLength				512
#define MaxWriteBlock			128
#define MaxServer				10

/* The multicast group for Replicate Files is 224.1.1.1 */
#define RFGROUP       			0xe0010101
#define	RFPORT					44015

/* Transimission parameter */
#define 	MAX_TRY 			3
#define 	TIMEOUT 			150

/* Packet types */
#define		PACKET_OPEN			0x01
#define		PACKET_ACK			0x02
#define		PACKET_NACK			0x03
#define		PACKET_WB			0x04
#define		PACKET_COMMIT		0x05
#define 	PACKET_COMMITNOW	0x06
#define		PACKET_ABORT		0x07
#define		PACKET_REQDATA		0x08
#define 	PACKET_CLOSE		0x09

/* Event type */
#define		EVENT_NETWORK		0x10
#define		EVENT_TIMEOUT		0x11

typedef	struct {
	unsigned int	messageID;			/* Message ID of the packet */
	unsigned int 	senderIP;			/* Sender's IP address */
	unsigned int	receiverIP;			/* Receiver's IP address, used to mimic unicast */
	unsigned int 	type;				/* The type/descriptor header of the packet */
	int				fd;					/* File descriptor */
	unsigned int	hash;				/* Security hash code, checksum */
	unsigned int	body[256];			/* The packet payload */
}	__attribute__ ((__packed__))	RF244BPacket;

typedef	struct {
	short			eventType;
	RF244BPacket	*eventDetail;		/* for incoming data */
	Sockaddr		eventSource;
}									RFEvent;

typedef struct {
	char			name[128];			/* The file name */
}	__attribute__ ((__packed__))	POPEN;

typedef struct {
	int 			blockSize;			/* Number of bytes of the block */
	int 			byteOffset;			/* Byte offset */	
	char 			buffer[MaxWBLength];		/* Content of the block */
}	__attribute__ ((__packed__))	PWriteBlock;

typedef struct {
	int				num;				/* The number of the write block messages */
}	__attribute__ ((__packed__))	PCommit;

typedef struct {
	int				messageID;			/* The messageID of the write block */
}	__attribute__ ((__packed__))	PReqData;

typedef struct {
	char			ID[128];
	unsigned int	IP;
} 									ServerType;

typedef struct {
	int				size;
	bool			ack[MaxServer];
	int				tries[MaxServer];
	int				timestamp[MaxServer];
} 	ACKQueue;

ServerType		servers[MaxServer];	/* The servers				*/
PWriteBlock		WBBuffer[MaxWriteBlock]; /* Write block buffer 	*/
int				WBBufferSize;		/* The size of the buffer	*/
int 			fileDescriptor;		/* Current file descriptor	*/
unsigned int 	messageID;			/* Current message ID		*/

unsigned int	myRFPort;			/* The port number 			*/
unsigned int	myIP;				/* My IP address			*/
Sockaddr 		*myAddr;			/* My socket address		*/
Sockaddr 		groupAddr;			/* Group Address used for SendData()	*/
int				mySocket;			/* the Socket used for communication	*/

int				serverNum;			/* The number of servers 	*/
int				packetLossRate;		/* The packet loss parameter */

int 			TimeZero;			/* The baseline time */

/* ------------------------------------------------------------------ */
int getCurrentTime (void);
int getServerIndex (unsigned int IP);
void initACKQueue (ACKQueue* queue);
int getNextTimeOut (ACKQueue queue);
Sockaddr* resolveHost(register char *name);
void RFError(char *s);
void RFWarn(char *s);
void NextEvent(RFEvent* event, int socket, int time);
void netInit(unsigned short portNum);

void sendOpenMsg (unsigned int receiverIP, int fd, char* fileName);
void sendWBMsg (unsigned int receiverIP, int idx, int fd);
void sendCommitMsg (unsigned int receiverIP, int size, int fd);
void sendCommitNowMsg (unsigned int receiverIP, int fd);
void sendAbortMsg (unsigned int receiverIP, int fd);
void sendCloseMsg(unsigned int receiverIP, int fd);
void sendPacket (RF244BPacket* pack);
void ConvertIncoming (RF244BPacket *pack);
unsigned int hashFunc(RF244BPacket* pack);
/* ------------------------------------------------------------------ */

int
InitReplFs( unsigned short portNum, int packetLoss ) {
#ifdef DEBUG
  	printf( "InitReplFs: Port number %d, packet loss %d percent\n", 
		  portNum, packetLoss );
#endif

	/****************************************************/
	/* Initialize network access, local state, etc.     */
  	/****************************************************/
  	fileDescriptor = 1;
 	messageID = 10000;
 	
 	serverNum = 0;
	packetLossRate = packetLoss;
	
	netInit(portNum);
  
	struct timeval timenow;
	gettimeofday(&timenow, 0);
	TimeZero = (timenow.tv_sec * 1000 + timenow.tv_usec / 1000);
 
	return( NormalReturn );  
}

/* ------------------------------------------------------------------ */

int
AddServer( char * id ) {
	Sockaddr	*thisHost;

#ifdef DEBUG
  printf( "AddServer: Server id %s\n", id );
#endif

	/****************************************************/
    /* Add server id to the list of servers             */
	/****************************************************/
	if ((thisHost = resolveHost(id)) == (Sockaddr *) NULL) {
		RFError("Cannot recognize server ID.");
	}
	
	servers[serverNum].IP = thisHost->sin_addr.s_addr;
	memcpy(servers[serverNum].ID, id, sizeof(id));
	serverNum ++;

#ifdef DEBUG
  printf( "AddServer: Server IP is %ld\n", servers[serverNum-1].IP );
#endif

	return( NormalReturn );  
}
/* ------------------------------------------------------------------ */

int
OpenFile( char * fileName ) {
	RF244BPacket*	pack;
	RFEvent			event;
	int fd;
	if (fileName == NULL) {
		RFWarn("Null file pointer.");
		return (ErrorReturn);
	}
//	ASSERT( fileName );

#ifdef DEBUG
	printf( "OpenFile: Opening File '%s'\n", fileName );
#endif
	
//	fd = open( fileName, O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR );

	/* Initialize write block buffer and file descriptor */
	WBBufferSize = 0;
	fd = (++fileDescriptor);
	
	/* Initialize ACK queue */
	ACKQueue 	queue;
	initACKQueue (&queue);
	
	messageID ++;
	for (int i = 0; i < serverNum; i++) {
		sendOpenMsg (servers[i].IP, fd, fileName);		
	}

	int nextTimeOut;
	while (1) {
		nextTimeOut = getNextTimeOut(queue);
		
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {			
			/* Check receiver IP */
			if (pack->receiverIP != myIP) 
				continue;
				
			/* This packet must have correct messageID */
			if (pack->messageID != messageID) {
				RFWarn("Received Wrong messageID.");
				continue;			
			}

			/* Check if the packet is an NACK */
			if (pack->type == PACKET_NACK) {
				RFWarn("Server has problem creating files");
				return (ErrorReturn);
			}

			/* Check if the packet is an ACK*/
			if (pack->type == PACKET_ACK) {
				int idx = getServerIndex(pack->senderIP);
				queue.ack[idx] = TRUE;
			}
		}
		
		/* Check whether we have received all the ACKs, and check if
		   there are any timeout requests */
		bool hasDone = TRUE;
		for (int i = 0; i < queue.size; i++) {
			if (queue.ack[i] == FALSE) {
				hasDone = FALSE;
				if (queue.timestamp[i] <= getCurrentTime()) {
					queue.tries[i] ++;
					if (queue.tries[i] > MAX_TRY) {
						RFWarn("One server is down");
						return (ErrorReturn);
					}
					sendOpenMsg (servers[i].IP, fd, fileName);
					queue.timestamp[i] = getCurrentTime() + TIMEOUT;
				}				
			}	
		}	
		if (hasDone) 
			break;
	}
	
#ifdef DEBUG
	if ( fd < 0 )
    	perror( "OpenFile" );
#endif
	return( fd );
}

/* ------------------------------------------------------------------ */

int
WriteBlock( int fd, char * buffer, int byteOffset, int blockSize ) {
	if (fd < 0) {
		RFWarn("Cannot handle negative fd!");
		return (ErrorReturn);
	}
//	ASSERT( fd >= 0 );
	if ( byteOffset < 0) {
		RFWarn("The byteoffset is negative.");
		return (ErrorReturn);
	}
//	ASSERT( byteOffset >= 0 );
	if (buffer == NULL) {
		RFWarn("Empty buffer data.");
		return (ErrorReturn);
	}
//	ASSERT( buffer );
	if ((blockSize < 0) || (blockSize >= MaxWBLength)) {
		RFWarn("Wrong block size!");
		return (ErrorReturn);
	}
//	ASSERT( blockSize >= 0 && blockSize < MaxWBLength );

	if (WBBufferSize >= MaxWriteBlock) {
		RFWarn("Maximum write block size reached!");
		return (ErrorReturn);
	}

#ifdef DEBUG
  	printf( "WriteBlock: Writing FD=%d, Offset=%d, Length=%d\n",
		fd, byteOffset, blockSize );
#endif

	if (fd != fileDescriptor) {
		RFWarn("Cannot write to a different file");
		return (ErrorReturn);
	}
	
	WBBuffer[WBBufferSize].blockSize = blockSize;
	WBBuffer[WBBufferSize].byteOffset = byteOffset;
	memcpy(WBBuffer[WBBufferSize].buffer, buffer, sizeof (buffer));
	
	sendWBMsg(0, WBBufferSize, fd);
	WBBufferSize ++;

/*	if ( lseek( fd, byteOffset, SEEK_SET ) < 0 ) {
    	perror( "WriteBlock Seek" );
    	return(ErrorReturn);
  	}

	if ( ( bytesWritten = write( fd, buffer, blockSize ) ) < 0 ) {
    	perror( "WriteBlock write" );
	    return(ErrorReturn);
  	}
*/
	return( blockSize );
}

/* ------------------------------------------------------------------ */

int
Commit( int fd ) {
	RF244BPacket*	pack;
	RFEvent		event;
	
	if (fd < 0) {
		RFWarn("File descriptor cannot be negative.");
		return (ErrorReturn);
	}
//	ASSERT( fd >= 0 );	

	if (fd != fileDescriptor) {
		RFWarn("Cannot commit another file");
		return (ErrorReturn);
	}

#ifdef DEBUG
	printf( "Commit: FD=%d\n", fd );
#endif

	bool success = TRUE;
	/****************************************************/
	/* Prepare to Commit Phase			    */
	/* - Check that all writes made it to the server(s) */
	/****************************************************/
	ACKQueue queue;
	initACKQueue (&queue);
	
	messageID ++;
	sendCommitMsg(0, WBBufferSize, fd);
	
	int nextTimeOut;
	while (1) {		
		nextTimeOut = getNextTimeOut(queue);
				
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {			
			/* Check receiver IP */
			if (pack->receiverIP != myIP) 
				continue;
			
			/* Check if the packet is an NACK */
			if (pack->type == PACKET_NACK) {
				RFWarn("One server has problem committing");
				success = FALSE;
				break;
			}
				
			/* This packet must be an ACK or ReqData */
			if (pack->type == PACKET_ACK) {					
				if (pack->messageID != messageID) {
					RFWarn("Received outdated ACK.");
					continue;			
				}
				int idx = getServerIndex(pack->senderIP);
				queue.ack[idx] = TRUE;
			} else if (pack->type == PACKET_REQDATA) {				
				int idx = getServerIndex(pack->senderIP);
				queue.timestamp[idx] = getCurrentTime() + TIMEOUT;
				sendWBMsg(servers[idx].IP, pack->messageID, fd);
			}
		}
		
		/* Check whether we have received all the ACKs, and check if
		   there are any timeout requests */
		bool hasDone = TRUE;
		for (int i = 0; i < queue.size; i++) {
			if (queue.ack[i] == FALSE) {
				hasDone = FALSE;
				if (queue.timestamp[i] <= getCurrentTime()) {
					queue.tries[i] ++;
					if (queue.tries[i] > MAX_TRY) {
						RFWarn("One server is down");
						success = FALSE;
						break;
					}
					sendCommitMsg(servers[i].IP, WBBufferSize, fd);
					queue.timestamp[i] = getCurrentTime() + TIMEOUT;
				}				
			}	
		}
		
		if (hasDone || !success) 
			break;
	}


	/****************/
	/* Commit Phase */
	/****************/
	
	if (!success) {
		RFWarn ("Commit fails");
		Abort(fd);
		return (ErrorReturn);
	}
	
	/* Do the real commit */
	initACKQueue (&queue);
	
	messageID ++;	
	sendCommitNowMsg(0, fd);
	
	while (1) {
		nextTimeOut = getNextTimeOut(queue);
		
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {
			
			/* Check receiver IP */
			if (pack->receiverIP != myIP) 
				continue;

			/* This packet must have correct messageID */
			if (pack->messageID != messageID) {
				RFWarn("Received Wrong messageID.");
				continue;			
			}

			/* This packet must be an ACK*/
			if (pack->type != PACKET_ACK) 
				continue;

			int idx = getServerIndex(pack->senderIP);
			queue.ack[idx] = TRUE;
		}
		
		/* Check whether we have received all the ACKs, and check if
		   there are any timeout requests */
		bool hasDone = TRUE;
		for (int i = 0; i < queue.size; i++) {
			if (queue.ack[i] == FALSE) {
				hasDone = FALSE;
				if (queue.timestamp[i] <= getCurrentTime()) {
					queue.tries[i] ++;
					if (queue.tries[i] > MAX_TRY) {
						RFWarn("Server is down on commitNow.");
						return (ErrorReturn);
					}
					sendCommitNowMsg(servers[i].IP, fd);
					queue.timestamp[i] = getCurrentTime() + TIMEOUT;
				}				
			}	
		}	
		if (hasDone) 
			break;
	}
	
	WBBufferSize = 0;
	
	return( NormalReturn );
}

/* ------------------------------------------------------------------ */

int
Abort( int fd ) {
	RF244BPacket*	pack;
	RFEvent		event;
	
	if (fd < 0) {
		RFError("File descriptor cannot be negative!");
	}
//	ASSERT( fd >= 0 );

	if (fd != fileDescriptor) {
		RFWarn("Cannot abort another file");
		return (ErrorReturn);
	}

#ifdef DEBUG
	printf( "Abort: FD=%d\n", fd );
#endif

	/*************************/
	/* Abort the transaction */
	/*************************/
	
	ACKQueue queue;
	initACKQueue (&queue);
	
	messageID ++;	
	sendAbortMsg(0, fd);
	
	int nextTimeOut;
	while (1) {
		nextTimeOut = getNextTimeOut(queue);
		
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {
			
			/* Check receiver IP */
			if (pack->receiverIP != myIP) 
				continue;
				
			/* This packet must have correct messageID */
			if (pack->messageID != messageID) {
				RFWarn("Received Wrong messageID.");
				continue;			
			}

			/* This packet must be an ACK*/
			if (pack->type != PACKET_ACK) 
				continue;
							
			int idx = getServerIndex(pack->senderIP);
			queue.ack[idx] = TRUE;
		}
		
		/* Check whether we have received all the ACKs, and check if
		   there are any timeout requests */
		bool hasDone = TRUE;
		for (int i = 0; i < queue.size; i++) {
			if (queue.ack[i] == FALSE) {
				hasDone = FALSE;
				if (queue.timestamp[i] <= getCurrentTime()) {
					queue.tries[i] ++;
					if (queue.tries[i] > MAX_TRY) {
						RFWarn("Server is down on abort.");
						return (ErrorReturn);
					}
					sendAbortMsg(servers[i].IP, fd);
					queue.timestamp[i] = getCurrentTime() + TIMEOUT;
				}				
			}	
		}	
		if (hasDone) 
			break;
	}	

	WBBufferSize = 0;
	
	return( NormalReturn );
}

/* ------------------------------------------------------------------ */

int
CloseFile( int fd ) {
	RF244BPacket*	pack;
	RFEvent		event;
	
	if (fd < 0) {
		RFWarn("File descriptor cannot be negative");
		return (ErrorReturn);
	}

	if (fd != fileDescriptor) {
		RFWarn("Cannot close another file");
		return (ErrorReturn);
	}
	
//	ASSERT( fd >= 0 );

#ifdef DEBUG
	printf( "Close: FD=%d\n", fd );
#endif

	/*****************************/
	/* Check for Commit or Abort */
	/*****************************/

	/* Attempt to commit before close */
	if (WBBufferSize > 0) {
		if (Commit (fd) < 0) {
			RFWarn("Commit before close file fails");
		}
	}
	
	
	/* Close file */
	ACKQueue queue;
	initACKQueue (&queue);
	
	messageID ++;	
	sendCloseMsg(0, fd);
	
	int nextTimeOut;
	while (1) {
		nextTimeOut = getNextTimeOut(queue);
		
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {
			
			/* Check receiver IP */
			if (pack->receiverIP != myIP) 
				continue;
				
			/* This packet must have correct messageID */
			if (pack->messageID != messageID) {
				RFWarn("Received Wrong messageID.");
				continue;			
			}
			
			/* Check if this packet is NACK */
			if (pack->type == PACKET_NACK) {
				RFWarn("One server encounters problem when closing file.");
				return (ErrorReturn);
			}			

			/* This packet must be an ACK*/
			if (pack->type != PACKET_ACK) 
				continue;	

			int idx = getServerIndex(pack->senderIP);
			queue.ack[idx] = TRUE;
		}
		
		/* Check whether we have received all the ACKs, and check if
		   there are any timeout requests */
		bool hasDone = TRUE;
		for (int i = 0; i < queue.size; i++) {
			if (queue.ack[i] == FALSE) {
				hasDone = FALSE;
				if (queue.timestamp[i] <= getCurrentTime()) {
					queue.tries[i] ++;
					if (queue.tries[i] > MAX_TRY) {
						RFWarn("Server is down on close.");
						return (ErrorReturn);
					}
					sendCloseMsg(servers[i].IP, fd);
					queue.timestamp[i] = getCurrentTime() + TIMEOUT;
				}				
			}	
		}	
		if (hasDone) 
			break;
	}	


//	if ( close( fd ) < 0 ) {
//    	perror("Close");
//	    return(ErrorReturn);
// 	}

	return(NormalReturn);
}

/* ------------------------------------------------------------------ */
void
netInit(unsigned short portNum)
{
	Sockaddr	nullAddr;
	Sockaddr	*thisHost;
	char		buf[128];
	int			reuse;
	u_char      ttl;
	struct ip_mreq  mreq;

	myRFPort = htons(portNum);

	gethostname(buf, sizeof(buf));	
	if ((thisHost = resolveHost(buf)) == (Sockaddr *) NULL)
		RFError("who am I?");
		
	myAddr = (Sockaddr*) malloc(sizeof(Sockaddr));	
	bcopy((caddr_t) thisHost, (caddr_t) (myAddr), sizeof(Sockaddr));
	
	/* Record my Ip address and port number */
	myIP = thisHost->sin_addr.s_addr;

	mySocket = socket(AF_INET, SOCK_DGRAM, 0);
	if (mySocket < 0)
		RFError("can't get socket");

	/* SO_REUSEADDR allows more than one binding to the same
	   socket */
	reuse = 1;
	if (setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
		RFError("setsockopt failed (SO_REUSEADDR)");
	}

	nullAddr.sin_family = AF_INET;
	nullAddr.sin_addr.s_addr = htonl(INADDR_ANY);
	nullAddr.sin_port = myRFPort;
	if (bind(mySocket, (struct sockaddr *)&nullAddr, sizeof(nullAddr)) < 0)
		RFError("netInit binding");

	/* Multicast TTL:
	   0 restricted to the same host
	   1 restricted to the same subnet
	   32 restricted to the same site
	   64 restricted to the same region
	   128 restricted to the same continent
	   255 unrestricted

	   DO NOT use a value > 32. If possible, use a value of 1 when
	   testing.
	*/

	ttl = 32;
	if (setsockopt(mySocket, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0) {
		RFError("setsockopt failed (IP_MULTICAST_TTL)");
	}

	/* Join the multicast group */
	mreq.imr_multiaddr.s_addr = htonl(RFGROUP);
	mreq.imr_interface.s_addr = htonl(INADDR_ANY);
	if (setsockopt(mySocket, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *) &mreq, sizeof(mreq)) < 0) {
		RFError("setsockopt failed (IP_ADD_MEMBERSHIP)");
	}

	/* Get the multi-cast address ready to use in SendData()
           calls. */
	memcpy(&groupAddr, &nullAddr, sizeof(Sockaddr));
	groupAddr.sin_addr.s_addr = htonl(RFGROUP);
}

void NextEvent(RFEvent *event, int socket, int time)
{
	fd_set	fdmask;
	struct timeval timeout;
	int		ret;
    
    if (time < 0)
    	time = 0;
    	
  	while (1) 
  	{
	  	FD_ZERO(&fdmask);
	  	FD_SET(socket, &fdmask);
	  	for (ret = 0; ret <= 0;) 
	  	{
	      	timeout.tv_sec = time / 1000;
	      	timeout.tv_usec = (time % 1000) * 1000;
	      	ret = select (32, &fdmask, NULL, NULL, &timeout);
	      
	      	if (ret == -1)
	      	{
			  	if (errno!=EINTR)
			  		RFError("select error on events");
		  	} else if (ret == 0) {
		    	event->eventType = EVENT_TIMEOUT;
   			  	return;
		  	}
	  	}

      	if (FD_ISSET(socket, &fdmask)) 
      	{
	  		socklen_t fromLen = sizeof(event->eventSource);
	  		int cc;
	  		event->eventType = EVENT_NETWORK;
	  		event->eventDetail = new RF244BPacket;	  		
	  		cc = recvfrom(socket, (char*)event->eventDetail, sizeof(RF244BPacket), 0,
		        (struct sockaddr *) &event->eventSource, &fromLen);
	 	 	if (cc <= 0)
	    	{
	      		if ((cc < 0) && (errno != EINTR))
					perror("event recvfrom");
	      		continue;
	    	}
	  		if (fromLen != sizeof(struct sockaddr_in))
	    		continue;
	  		ConvertIncoming(event->eventDetail);
	  		return;
		}
    }
}


/*------------------    Utility functions   ----------------*/
/* Get current time */
int getCurrentTime () {
	struct timeval timenow;
	gettimeofday(&timenow, 0);
	return (timenow.tv_sec * 1000 + timenow.tv_usec / 1000 - TimeZero);

}

/* Get server index by IP */
int getServerIndex (unsigned int IP) {
	int ret = -1;
	for (int i = 0; i < serverNum; i++) 
		if (servers[i].IP == IP)
			ret = i;
	if (ret < 0)
		RFWarn("Cannot find server IP");
	return ret;
}

/* Initilize ACK queue */
void initACKQueue (ACKQueue *queue) {
	queue->size = serverNum;
	for (int i = 0; i < serverNum; i++) {
		queue->ack[i] = FALSE;
		queue->tries[i] = 0;
		queue->timestamp[i] = getCurrentTime() + TIMEOUT;
	}
	return;
}

/* Get next timeout number */
int getNextTimeOut (ACKQueue queue) {
	int	nextTimeOut = getCurrentTime();
	for (int i = 0; i < serverNum; i++) {
		if ((queue.ack[i] == FALSE) && (queue.timestamp[i] > nextTimeOut)) 
			nextTimeOut = queue.timestamp[i];
	}
	return nextTimeOut;
}


/*
 * Resolve the specified host name into an internet address.  The "name" may
 * be either a character string name, or an address in the form a.b.c.d where
 * the pieces are octal, decimal, or hex numbers.  Returns a pointer to a
 * sockaddr_in struct (note this structure is statically allocated and must
 * be copied), or NULL if the name is unknown.
 */

Sockaddr *
resolveHost(register char *name)
{
	register struct hostent *fhost;
	struct in_addr fadd;
	static Sockaddr sa;

	if ((fhost = gethostbyname(name)) != NULL) {
		sa.sin_family = fhost->h_addrtype;
		sa.sin_port = 0;
		bcopy(fhost->h_addr, &sa.sin_addr, fhost->h_length);
	} else {
		fadd.s_addr = inet_addr(name);
		if (fadd.s_addr != -1) {
			sa.sin_family = AF_INET;	/* grot */
			sa.sin_port = 0;
			sa.sin_addr.s_addr = fadd.s_addr;
		} else
			return(NULL);
	}
	return(&sa);
}

/* Output error message */
void RFError(char *s)
{
	fprintf(stderr, "CS244B Distributed Replicated Files: %s\n", s);
	perror("CS244B Assignment 2");
	exit(-1);
}

/* Output warning message */
void RFWarn(char *s)
{
	fprintf(stderr, "WARNING: Distributed Replicated Files: %s\n", s);
	return;
}

/* Send "Open" message to specified server */
void sendOpenMsg (unsigned int receiverIP, int fd, char *fileName) {
printf("Open : receiver = %ld, fd = %d, name = %s\n", receiverIP, fd, fileName);
	RF244BPacket *pack = new RF244BPacket;
	POPEN* packX = (POPEN *) &(pack->body);
	memset(packX, 0, 256 * sizeof(unsigned int));
	 
	pack->type = PACKET_OPEN;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	strcpy(packX->name, fileName);
	
	sendPacket(pack);
	free(pack);
}

void sendWBMsg (unsigned int receiverIP, int idx, int fd) {
printf("SendWB : receiver = %ld, fd = %d, idx = %d\n", receiverIP, fd, idx);
	RF244BPacket *pack = new RF244BPacket;
	PWriteBlock* packX = (PWriteBlock *) &(pack->body);
	memset(packX, 0, 256 * sizeof(unsigned int));	
	
	pack->messageID = idx;
	pack->type = PACKET_WB;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	
	packX->blockSize = htonl(WBBuffer[idx].blockSize);
	packX->byteOffset = htonl(WBBuffer[idx].byteOffset);
	memcpy(packX->buffer, WBBuffer[idx].buffer, WBBuffer[idx].blockSize);
	sendPacket(pack);	
	free(pack);	
}

void sendCommitMsg (unsigned int receiverIP, int size, int fd) {
printf("Commit : receiver = %d, fd = %d, size = %d\n", receiverIP, fd, size);
	RF244BPacket *pack = new RF244BPacket;
	PCommit* packX = (PCommit *) &(pack->body);
	memset(packX, 0, 256 * sizeof(unsigned int));	
	
	pack->type = PACKET_COMMIT;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	
	packX->num = htonl(size);
	sendPacket(pack);	
	free(pack);	
}

void sendCommitNowMsg (unsigned int receiverIP, int fd) {
printf("CommitNow : receiver = %d, fd = %d\n", receiverIP, fd);
	//FIXME consider adding some delay here

	RF244BPacket *pack = new RF244BPacket;
	
	pack->type = PACKET_COMMITNOW;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	
	sendPacket(pack);	
	free(pack);	
}

void sendAbortMsg (unsigned int receiverIP, int fd) {
printf("Abort : receiver = %d, fd = %d\n", receiverIP, fd);
	//FIXME consider adding some delay here
	RF244BPacket *pack = new RF244BPacket;
	
	pack->type = PACKET_ABORT;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	
	sendPacket(pack);	
	free(pack);	
}

void sendCloseMsg(unsigned int receiverIP, int fd) {
printf("Close : receiver = %d, fd = %d\n", receiverIP, fd);
	RF244BPacket *pack = new RF244BPacket;
	
	pack->type = PACKET_CLOSE;
	pack->receiverIP = receiverIP;
	pack->fd = fd;
	
	sendPacket(pack);
	free(pack);	
}

void sendPacket (RF244BPacket *pack) {
	/* Simulating packet loss */
	if (rand() % 100 < packetLossRate)
		return;
	
	if (pack->type != PACKET_WB)
		pack->messageID = messageID;
	pack->senderIP = myIP;
	pack->hash = hashFunc(pack);
	
	pack->messageID = htonl(pack->messageID);
	pack->senderIP = htonl(pack->senderIP);
	pack->receiverIP = htonl(pack->receiverIP);
	pack->type = htonl(pack->type);
	pack->fd = htonl(pack->fd);
	pack->hash = htonl(pack->hash);		

	if (sendto((int)mySocket, pack, sizeof(*pack), 0,
		   (struct sockaddr*) &groupAddr, sizeof(Sockaddr)) < 0)
	  { RFError("Send packet Error"); }	
}

void ConvertIncoming (RF244BPacket *pack) {
	pack->messageID = ntohl(pack->messageID);
	pack->senderIP	= ntohl(pack->senderIP);
	pack->receiverIP = ntohl(pack->receiverIP);
	pack->type = ntohl(pack->type);
	pack->fd = ntohl(pack->fd);
	pack->hash = ntohl(pack->hash);
	if (pack->hash != hashFunc(pack)) {
		RFWarn("Checksum failed, the packet is discarded");
		pack->receiverIP = 1;
		return;
	}
	if (pack->fd != fileDescriptor) {
		RFWarn("Received packet with a different fd, discard.");
		pack->receiverIP = 1;
		return;
	}
}

unsigned int hashFunc(RF244BPacket *pack) {
	unsigned int value = 0;
	for (int i = 0; i < 256; i++)
		value = value ^ pack->body[i];
	value = value ^ pack->messageID ^ pack->senderIP ^ pack->receiverIP ^ pack->type ^ pack->fd;
	return value;
}
